package com.dmp.storm.topology.db;


import com.dmp.storm.topology.BaseTopologyCtx;

import java.sql.*;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class DbTopologyCtx extends BaseTopologyCtx {
	private Connection conn;

	public DbTopologyCtx(Properties props, String... tables)
			throws SQLException {
		super();
		try {
			conn = getConn(props.getProperty("url"),
					props.getProperty("username"),
					props.getProperty("password"));
			for (String table : tables) {
				createWorkFlow(SS.LOCAL, table);
			}
		} finally {
			conn.close();
		}
	}

	protected Connection getConn(String url, String user, String password)
			throws SQLException {
		return DriverManager.getConnection(url, user, password);
	}

	private Map<String, String> getComponent(ResultSet rs) {
		Map<String, String> component = new HashMap<String, String>();
		try {
			component.put("name", rs.getString("name"));
			component.put("type", rs.getString("type"));
			component.put("class", rs.getString("class"));
			component.put("parallenum", rs.getString("parallenum"));
			component.put("groupname", rs.getString("groupname"));
			component.put("grouptype", rs.getString("grouptype"));
		} catch (SQLException e) {
			System.err.println("获得组件时发生异常");
			e.printStackTrace();
			System.exit(-1);
		}
		return component;
	}

	private Map<String, String> getComponentProp(ResultSet rs)
			throws SQLException {
		Map<String, String> propsMap = new HashMap<String, String>();
		while (rs.next()) {
			propsMap.put(rs.getString("attrkey"), rs.getString("attrvalue"));
		}
		return propsMap;
	}

	@Override
	protected void createWorkFlow(BaseTopologyCtx.SS ss, String tablename) {
		try {
			PreparedStatement componentPsmt = conn
					.prepareStatement("select * from " + tablename);
			ResultSet componentRs = componentPsmt.executeQuery();
			while (componentRs.next()) {
				Map<String, String> components = this.getComponent(componentRs);
				String name = components.get("name");
				String type = components.get("type");
				String classPath = components.get("class");
				String num = components.get("parallenum");
				PreparedStatement componentPropPsmt = conn
						.prepareStatement("select * from " + tablename
								+ "prop where name=?");
				componentPropPsmt.setString(1, name);
				ResultSet componentPropRs = componentPropPsmt.executeQuery();
				Map<String, String> props = this
						.getComponentProp(componentPropRs);
				if ("group".equals(type)) {
					this.createStreamGrouping(name, classPath, props);
				}
				if ("spout".equals(type)) {
					this.createSpout(name, classPath, props, num);
				}
				if ("bolt".equals(type)) {
					this.createBolt(name, classPath, props, num,
							components.get("grouptype"),
							components.get("groupname"));
				}
			}
		} catch (SQLException e) {
			System.err.println("创建工作流发生异常");
			e.printStackTrace();
		} finally {
			try {
				conn.close();
				System.exit(-1);
			} catch (SQLException e) {
				e.printStackTrace();
				System.exit(-1);
			}
		}
	}
}
